//----------------------------------------------------------
    // 异步工作区构建
        // 1.异步工作线程的创建
        // 2.缓冲区的实现
//----------------------------------------------------------

#ifndef __LOGGER_ASYNCTHREAD__
#define __LOGGER_ASYNCTHREAD__

#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <vector>
#include <string>
#include <deque>
#include <thread>

namespace ymbxlog
{
    #define BUFFER_BASE_SIZE 1024*1024
    #define BUFFER_THRESHOLD_SIZE 5*1024*1024
    #define BUFFER_LINER_INCRE_SIZE 1*1024*1024
        class Buffer{
        public:
            Buffer():_rd_loc(0),_wr_loc(0){
                _buf.resize(BUFFER_BASE_SIZE);
            }
            void Push(const std::string& instr){
                if(_wr_loc+instr.size()>_buf.size()){
                    ExpandSize(_wr_loc+instr.size());
                }
                std::copy(instr.begin(),instr.end(),_buf.begin()+_wr_loc);
                _wr_loc+=instr.size();
            }
            // 返回读取到的字节数
            size_t Pop(size_t size){
                if(_rd_loc+size>_wr_loc){
                    return _wr_loc-_rd_loc;
                }
                return size;
            }
            void Swap(Buffer& buf){
                std::swap(_buf,buf._buf);
                std::swap(_rd_loc,buf._rd_loc);
                std::swap(_wr_loc,buf._wr_loc);
            }
            size_t WriteAbleSize(){
                return _buf.size()-_wr_loc;
            }
            size_t ReadAbleSize(){
                return _wr_loc-_rd_loc;
            }
            bool IsEmpty(){
                return _rd_loc==_wr_loc;
            }
            void Reset(){
                _rd_loc=0;
                _wr_loc=0;
            }
            size_t Size(){
                return _buf.size();
            }
            std::string RetStr(){
                std::string tmp;
                for(int i=_rd_loc;i<_wr_loc;++i){
                    tmp+=_buf[i];
                }
                return tmp;
            }
        private:
            void ExpandSize(size_t need_size){
                while(need_size>_buf.size()){
                    if(_buf.size()<BUFFER_THRESHOLD_SIZE){
                        _buf.resize(_buf.size()*2);
                    }else{
                        _buf.resize(_buf.size()+BUFFER_LINER_INCRE_SIZE);
                    }
                }
            }
        private:
            size_t _rd_loc;
            size_t _wr_loc;
            std::vector<char> _buf;
        };
        class AsyncLooper
        {
        public:
            using Functor = std::function<void(Buffer &)>;
            using ptr = std::shared_ptr<AsyncLooper>;
            AsyncLooper(size_t thread_cnt,const Functor &cb) : _callback(cb), _run(true){
                for(int i=0;i<thread_cnt;++i){
                    _threads.emplace_back(&AsyncLooper::ThreadEntry,this);
                }
            }
            ~AsyncLooper()
            {
                Stop();
            }
            void Stop()
            {
                _run.store(false);
                _cons_conv.notify_all();
                _prod_conv.notify_all();
                for(auto& t:_threads){
                    if(t.joinable()){
                        t.join();
                    }
                }
            }
            void Push(const std::string& mes)
            {
                if (_run == false)
                    return;
                {
                    std::unique_lock<std::mutex> lock(_wr_mtx);

                    _prod_conv.wait(lock, [&]()
                                      { return !_run || _push_buffer.WriteAbleSize() >= mes.size(); });
                    if(!_run) return;
                    _push_buffer.Push(mes);
                }
                _cons_conv.notify_all();
            }
    
        private:
            // 线程入口函数
            void ThreadEntry()
            {
                while (1)
                {
                    {
                        std::unique_lock<std::mutex> lock(_rd_mtx);
                        //!!!!!!!!!!!!!!!!!!!!!!!!!
                        _cons_conv.wait(lock, [&]()
                                      { return !_run || !_push_buffer.IsEmpty(); });
                        if(!_run && _pop_buffer.IsEmpty() && _push_buffer.IsEmpty()) break;
                        if(!_push_buffer.IsEmpty())
                        {
                            std::unique_lock<std::mutex> lck(_wr_mtx);
                            _push_buffer.Swap(_pop_buffer);
                        }
                        _prod_conv.notify_all();
                    }
                    _callback(_pop_buffer);
                    _pop_buffer.Reset();
                }
            }
    
        private:
            Functor _callback;
    
        private:
            Buffer _push_buffer;
            Buffer _pop_buffer;
            std::vector<std::thread> _threads;
            std::mutex _rd_mtx;
            std::mutex _wr_mtx;
            std::atomic<bool> _run; // 标志位，表示写入缓冲区线程是否工作————true表示工作
            std::condition_variable _prod_conv;
            std::condition_variable _cons_conv;
        };
}

#endif